package com.zxj.hadoop.demo.mapreduce.join;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author 朱小杰
 * 时间 2017-08-22 .22:10
 * 说明 ...
 */
public class MRJoin {
    /**
     * 这个类的信息，包含了两个表的信息记录
     */
    static class Info implements Writable,Cloneable{
        /**
         * 订单号
         */
        private int orderId;
        /**
         * 时间
         */
        private String dateString;
        /**
         * 产品编号
         */
        private String pid;
        /**
         * 数量
         */
        private int amount;
        /**
         * 产品名称
         */
        private String pname;
        /**
         * 种类
         */
        private int categoryId;
        /**
         * 价格
         */
        private float price;
        /**
         * 这个字段需要理解<br>
         * 因为这个对象，包含了订单与产品的两个文件的内容，当我们加载一个文件的时候，肯定只能加载一部分的信息，另一部分是加载不到的，需要在join的时候，加进去，这个字段就代表着这个对象存的是哪些信息
         * 如果为0  则是存了订单信息
         * 如果为1 则是存了产品信息
         */
        private String flag;

        @Override
        protected Object clone() throws CloneNotSupportedException {
            return super.clone();
        }

        @Override
        public void write(DataOutput out) throws IOException {
            out.writeInt(orderId);
            out.writeUTF(dateString);
            out.writeUTF(pid);
            out.writeInt(amount);
            out.writeUTF(pname);
            out.writeInt(categoryId);
            out.writeFloat(price);
            out.writeUTF(flag);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            orderId = in.readInt();
            dateString = in.readUTF();
            pid = in.readUTF();
            amount = in.readInt();
            pname = in.readUTF();
            categoryId = in.readInt();
            price = in.readFloat();
            flag = in.readUTF();
        }

        public Info() {
        }

        public void set(int orderId, String dateString, String pid, int amount, String pname, int categoryId, float price,String flag) {
            this.orderId = orderId;
            this.dateString = dateString;
            this.pid = pid;
            this.amount = amount;
            this.pname = pname;
            this.categoryId = categoryId;
            this.price = price;
            this.flag = flag;
        }

        public int getOrderId() {
            return orderId;
        }

        public void setOrderId(int orderId) {
            this.orderId = orderId;
        }

        public String getDateString() {
            return dateString;
        }

        public String getFlag() {
            return flag;
        }

        public void setFlag(String flag) {
            this.flag = flag;
        }

        public void setDateString(String dateString) {
            this.dateString = dateString;
        }

        public String getPid() {
            return pid;
        }

        public void setPid(String pid) {
            this.pid = pid;
        }

        public int getAmount() {
            return amount;
        }

        public void setAmount(int amount) {
            this.amount = amount;
        }

        public String getPname() {
            return pname;
        }

        public void setPname(String pname) {
            this.pname = pname;
        }

        public int getCategoryId() {
            return categoryId;
        }

        public void setCategoryId(int categoryId) {
            this.categoryId = categoryId;
        }

        public float getPrice() {
            return price;
        }

        public void setPrice(float price) {
            this.price = price;
        }

        @Override
        public String toString() {
            final StringBuilder sb = new StringBuilder("{");
            sb.append("\"orderId\":")
                    .append(orderId);
            sb.append(",\"dateString\":\"")
                    .append(dateString).append('\"');
            sb.append(",\"pid\":")
                    .append(pid);
            sb.append(",\"amount\":")
                    .append(amount);
            sb.append(",\"pname\":\"")
                    .append(pname).append('\"');
            sb.append(",\"categoryId\":")
                    .append(categoryId);
            sb.append(",\"price\":")
                    .append(price);
            sb.append(",\"flag\":\"")
                    .append(flag).append('\"');
            sb.append('}');
            return sb.toString();
        }
    }

    static class JoinMapper extends Mapper<LongWritable,Text,Text,Info>{
        private Info info = new Info();
        private Text text = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if(line.startsWith("#")){//跳转带#的注释
                return;
            }
            //获取当前任务的输入切片，这个InputSplit是一个最上层抽象类，可以转换成FileSplit
            InputSplit inputSplit = context.getInputSplit();
            FileSplit fileSplit = (FileSplit) inputSplit;
            String name = fileSplit.getPath().getName();//得到的是文件名，这里根据文件名来判断是哪一种类型的数据
            //我们这里通过文件名判断是哪种数据
            String pid = "";
            String[] split = line.split(",");
            if(name.startsWith("order")){//加载订单内容，订单数据里面有 订单号，时间，产品ID，数量
                //orderId,date,pid,amount
                pid = split[2];
                info.set(Integer.parseInt(split[0]),split[1],pid,Integer.parseInt(split[3]),"",0,0,"0");

            }else{//加载产品内容，产品数据有 产品编号，产品名称，种类，价格
                //pid,pname,categoryId,price
                pid = split[0];
                info.set(0,"",pid,0,split[1],Integer.parseInt(split[2]),Float.parseFloat(split[3]),"1");
            }
            text.set(pid);
            context.write(text,info);
        }
    }

    static class JoinReduce extends Reducer<Text,Info,Info,NullWritable>{

        @Override
        protected void reduce(Text key, Iterable<Info> values, Context context) throws IOException, InterruptedException {
            Info product = new Info();//这个对象用来存放产品的数据，一个产品所以只有一个对象
            List<Info> infos = new ArrayList<>();//这个list用来存放所有的订单数据，订单肯定是有多个的
            for(Info info : values){
                if("1".equals(info.getFlag())){
                    //产品表的数据
                    try {
                        product = (Info) info.clone();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }else{//代表着是订单表的数据
                    Info order = new Info();
                    try {
                        order = (Info) info.clone();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    infos.add(order);
                }
            }
            //经过上面的操作，就把订单与产品完全分离出来了，订单在list集合中，产品在单独的一个对象中
            //然后可以分别综合设置进去
            for(Info tmp : infos){
                tmp.setPname(product.getPname());
                tmp.setCategoryId(product.getCategoryId());
                tmp.setPrice(product.getPrice());
                //最后进行输出，就会得到结果文件
                context.write(tmp,NullWritable.get());
            }
        }
    }


    static class JoinMain{
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);

            job.setJarByClass(JoinMain.class);

            job.setMapperClass(JoinMapper.class);
            job.setReducerClass(JoinReduce.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Info.class);

            job.setOutputKeyClass(Info.class);
            job.setOutputValueClass(NullWritable.class);

            FileInputFormat.setInputPaths(job,new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));

            boolean b = job.waitForCompletion(true);
            if(b){
                System.out.println("OK");
            }

        }
    }



}
